[event] add optimizer event stream publisher - #289

Writing to a TCP socket with ::write() can raise SIGPIPE if the peer has closed its half of the connection. Since the process does not appear to ignore SIGPIPE globally, a consumer disconnecting mid-send could terminate the publisher worker or IO thread. Use ::send(fd, ..., MSG_NOSIGNAL) here, or ensure SIGPIPE is ignored process-wide.

🤖 Generated by Qoder

该评论对应的专用 TCP 事件流已被移除，当前改为复用 KVCM 现有 gRPC Server 的 server-streaming 接口，不再直接调用 write，因此 SIGPIPE 问题已随旧实现淘汰。

Binding the optimizer event stream to INADDR_ANY exposes this port on all network interfaces. Because this is a diagnostic/replay channel, consider defaulting to INADDR_LOOPBACK or adding a bind_address config option so it is not reachable from untrusted networks by default.

🤖 Generated by Qoder

该评论对应的专用 TCP 监听端口已移除。事件流现在复用 KVCM 已有的 gRPC Server，不再额外绑定 INADDR_ANY，网络暴露范围沿用现有服务的部署与安全边界。

key is an int64_t, so the range-for currently copies each value. Use const auto &key to avoid the unnecessary copy.

🤖 Generated by Qoder

该评论所在代码已在后续重构中移除，当前 gRPC 事件转换实现不再包含这处 range-for 拷贝。

Capping queue_size at size_t::max() means a malformed config can request an impossibly large queue and cause bad_alloc during startup. Consider bounding this to a sane operational maximum (e.g., a few million).

🤖 Generated by Qoder

该 queue_size 属于已删除的 TCP 事件流配置。当前改为 gRPC 订阅队列，并使用独立的 subscriber_queue_size 配置，因此此处的超大 TCP 队列分配问题已不再存在。

Send() holds connections_mutex_ while it frames the payload and writes to every ready socket. This serializes the publisher worker with the IO loop and prevents it from processing handshakes/disconnects while a large frame is being pushed. Consider framing the payload outside the lock, copying the ready FDs, and writing without holding the mutex.

🤖 Generated by Qoder

该专用 TCP Sink 已被 gRPC server-streaming 实现替代，connections_mutex_、socket framing 和持锁写 socket 的路径均已删除，因此该锁竞争问题已随旧实现淘汰。

dropped_ is incremented once for every subscriber whose queue is full, and again when there are no subscribers at all. That means a single event can inflate dropped_ by up to max_subscribers, and Send() can return true while dropped_ grows. If this counter is meant to be compared against forwarded_, the per-subscriber semantics are surprising. Consider documenting it explicitly or counting a drop only when no subscriber accepted the event.

🤖 Generated by Qoder

这里按 best-effort 分析链路处理，允许少量事件丢失；dropped_ 表示订阅投递失败次数，不与 forwarded_ 做一一对应。当前已更新注释，避免把该计数解释为完整性保证，暂不引入更复杂的逐事件投递统计。

sink_->stopped() and sink_->Subscribe() are not atomic. If Stop() is called between the two, Subscribe() returns nullptr and the RPC reports RESOURCE_EXHAUSTED even though the sink is unavailable. Returning UNAVAILABLE when stopped_ is true would be more accurate; consider checking sink_->stopped() again after a failed subscribe.

🤖 Generated by Qoder

已修复。订阅创建与 Sink 的停止/禁用检查现在由同一把锁线性化；服务对外只暴露可用/不可用二态，订阅失败统一返回 gRPC UNAVAILABLE，不再把 Stop 竞态误报为 RESOURCE_EXHAUSTED。

GetConfiguration returns the KVCM topology without checking whether this node is the leader. If an optimizer connects to a standby, it may receive stale or empty configuration and no events. Consider either enforcing a leader check (consistent with other meta requests) or documenting that optimizers must target the leader.

🤖 Generated by Qoder

已修复。GetConfiguration 现在仅在节点处于稳定 Leader、Registry 恢复完成且事件服务已启用时返回配置；其他情况统一返回 SERVICE_NOT_READY，Optimizer 无需区分 KVCM 内部角色状态。

A malformed event_publishers_configs string disables both the log and optimizer publishers. If only the optimizer config is invalid, the log publisher is still taken down. Consider falling back to the log publisher when the optimizer section is the only invalid part, or at least logging which publisher is affected.

🤖 Generated by Qoder

部分采纳。当前配置仍是一个整体 JSON，整体解析失败时无法可靠判断仅哪个子配置无效，因此继续 fail closed 并记录明确错误；但各 Publisher 的初始化/注册已互相隔离，Log Publisher 注册失败不会再阻止 Optimizer Publisher 注册，失败对象也会立即停止。

Gate configuration snapshots on recovery readiness

During startup, and permanently on a follower, RegistryManager::ListInstanceGroup can legitimately return an empty or partially recovered in-memory registry, yet this method reports OK. Server::Start exposes the RPC before starting leader election, while registry recovery happens later in OnBecomeLeader; RegistryManager already exposes IsRecoverComplete() and the protocol defines SERVICE_NOT_READY/SERVER_NOT_LEADER. An optimizer that bootstraps in this window can accept an empty snapshot and then discard streamed events for unknown instances, so return a non-OK application status until the node is the recovered leader.

Useful? React with 👍 / 👎.

已修复。配置快照增加统一可用性门禁：只有稳定 Leader 且 Registry 恢复完成后才返回 OK；启动、Follower、晋升/降级和恢复阶段统一返回 SERVICE_NOT_READY，避免 Optimizer 接受空或未恢复完成的配置。

Preserve timestamp order across concurrent publishers

With concurrent successful cache reads for the same instance, one serving thread can call SetEventTriggerTime() first but reach BasicEnqueue() after another thread, so FIFO queue order can contain decreasing timestamps. A single worker preserves only enqueue order, not the claimed timestamp order, and the replay/TTL logic assumes non-decreasing timestamps; this can reorder accesses or evaluate expirations against time moving backward. Serialize or clamp ordering per instance before sending events.

Useful? React with 👍 / 👎.

确认该窗口存在，但本次不在 KVCM 侧通过 clamp 修改原始 timestamp，也不在在线请求链路引入排序锁或等待缓存。KVCM 保留 producer timestamp，并由单 worker 按入队 FIFO 发送；同一 Instance 按真实时间戳的有限窗口重排属于 Optimizer 消费侧处理。

Expose subscriber queue losses to the optimizer

When a subscriber is slower than subscriber_queue_size, this path silently drops the event and only increments an in-process counter that is neither included in the stream nor registered as a metric; it is otherwise logged only when the sink stops. The optimizer therefore continues producing capacity curves from an incomplete trace without knowing that samples were lost. Surface per-subscriber loss through stream metadata, sequence gaps, or operational metrics so affected results can be rejected.

Useful? React with 👍 / 👎.

按当前产品定位不做该复杂度改动。Optimizer 是 best-effort 理论分析链路，可接受少量事件丢失；这里不要求无损投递或由消费端拒绝整段结果，因此暂不增加序列号、stream metadata 或逐订阅指标。

Terminate subscriptions when leadership is lost

A follower accepts SubscribeEvents whenever the process-local sink is running, and an existing stream is not closed when OnNoLongerLeader disables leader-only cache reads. Consequently, after failover the optimizer can remain attached to the old leader on a healthy but permanently idle stream and never reconnect to the new event producer; the same occurs if it initially contacts a follower. Reject subscriptions unless the node is the recovered leader and close active subscriptions on demotion so clients receive an error and reconnect.

Useful? React with 👍 / 👎.

已修复。Follower/恢复中的节点统一拒绝订阅并返回 UNAVAILABLE；Leader 开始降级或 Server 停止时会先禁用新订阅并主动关闭已有流，使 Optimizer 及时感知不可用并重新发现 Leader、同步配置和建流。

Document the new event-to-protocol dependency

This target introduces a direct event → protocol module dependency, but neither the root dependency diagram nor docs/design/module_architecture.md was updated; the latter still depicts event as depending only on common. Update both architecture diagrams and the event module description so future dependency checks account for this new edge.

AGENTS.md reference: AGENTS.md:L38-L38

Useful? React with 👍 / 👎.

已修复。根目录 AGENTS.md 与 docs/design/module_architecture.md 的依赖图均补充了 event → protocol 边，并更新 event/protocol 模块说明，记录 OptimizerEventPublisher、TraceQueryRequest、SubscriptionEventSink 与 gRPC 事件流的关系。

Honor producer timestamps during optimizer replay

When TTL is enabled and an event spends time in the publisher, subscriber, or network queues, this timestamp currently has no effect: OptimizerServiceImpl::TraceQuery never reads request->timestamp_ns() and OnlineOptimizerManager::TraceQuery calls ProcessRequest with its current wall clock instead. Consequently, delayed events are replayed too late and TTL evictions and resulting capacity curves are incorrect despite the new field explicitly promising producer-time evaluation; pass this value through the replay path and only fall back to the wall clock when it is zero.

Useful? React with 👍 / 👎.

该问题发生在 Optimizer 消费/回放链路，不在本次 KVCM 修复范围。KVCM 已保留并传输 producer 的 timestamp_ns；Optimizer 后续应将该字段继续传给回放引擎，仅在值为 0 时使用本地时间兜底，避免影响 TTL 语义。

Build configuration under one registry lock

When instance registrations or removals run concurrently with this RPC, each ListInstanceInfo call acquires a fresh registry lock after ListInstanceGroup has already released its lock, so the returned configuration is not a point-in-time snapshot. For example, the RPC can list instance A, then an admin removes A and registers B in another group, and the subsequent iteration lists B, yielding a response containing A and B even though they never coexisted; bootstrapping the optimizer from that response leaves stale instance state. Capture groups and their instances together under one shared registry lock or expose an atomic registry snapshot API.

Useful? React with 👍 / 👎.

这里按 best-effort、最终一致的分析语义处理。Group/Instance 在管理变更期间出现短暂混合快照可以接受，稳定后重新获取配置即可纠正；为此新增 Registry 原子拓扑快照会扩大锁范围和接口复杂度，因此本次不改。

Do not replay batch lookups as prefix queries

When KVCM serves QT_BATCH_GET, this conversion forwards the arbitrary batch key list as a TraceQueryRequest even though that protocol defines block_keys as an ordered prefix and the online optimizer computes continuous prefix hits from it. Because the conversion also drops CacheGetEvent::query_type(), a batch such as hit/miss/hit is indistinguishable from a prompt prefix and is accounted with prefix-hit semantics, corrupting aggregate hit rates and capacity curves. Filter out non-prefix-compatible query types or carry their query semantics through to a matching replay implementation.

Useful? React with 👍 / 👎.

按 KVCM 职责边界不在这里过滤。KVCM 只透明采集和转发已发生的 Query，不理解或重解释 QT_BATCH_GET 等具体查询语义；如不同查询类型需要不同回放模型，应由协议语义和 Optimizer 消费实现统一处理。

Avoid logging every full optimizer enqueue

When the optimizer publisher queue reaches queue_size, BasicEnqueue returns false here, causing EventManager::Publish to issue an unthrottled KVCM_LOG_WARN for every cache read (event_manager.cc:114-117). Because this runs on serving threads and the logger serializes calls through its mutex, optimizer backpressure can create a warning storm and slow the request path that this bounded queue is intended to protect; count or rate-limit these drops without reporting each one as a publish failure.

Useful? React with 👍 / 👎.

已修复。队列满仍由 BasicEnqueue 统计 dropped，但 OptimizerEventPublisher 将这种预期的 best-effort 丢弃视为已处理，不再向 EventManager 返回失败，因此不会在请求线程为每个丢弃事件输出 WARN；未初始化或已停止仍返回失败。

Propagate live registry changes to subscribers

When an instance or group is registered, removed, or updated after the optimizer's initial GetConfiguration call, this stream can only deliver TraceQueryRequest messages, and the repository's CacheManager::RegisterInstance/RemoveInstance paths publish no corresponding lifecycle event. The optimizer therefore receives accesses for newly registered unknown instances and retains stale state for removed or reconfigured instances, which can corrupt shared-group capacity results until it reconnects; add versioned configuration/lifecycle messages or another mandatory resynchronization mechanism.

Useful? React with 👍 / 👎.

按当前 best-effort 分析语义，本次不增加版本化配置消息或生命周期协议。运行时注册变更允许造成短暂不一致，Optimizer 在连接重建时重新调用 GetConfiguration，稳定状态下即可收敛；若未来要求持续在线强一致，再单独设计配置版本与增量同步机制。

Revalidate leadership before returning the snapshot

When GetConfiguration passes the initial availability check immediately before demotion begins, Server::OnNoLongerLeader does not wait for this RPC before RegistryManager::DoCleanup clears the registry. The group query can therefore finish before cleanup while later instance queries run afterward, and this line still marks the resulting partial response as OK. Fresh evidence in this revision is that the new gate is checked only at entry, with no leader-only request guard or final availability check; keep the RPC in the demotion drain or discard the response if availability changes while it is being built.

Useful? React with 👍 / 👎.

已修复。GetConfiguration 在完成 Group/Instance 响应组装后会再次检查 KVCM 可用性；若期间发生降级、恢复状态变化或订阅被禁用，则清空已组装的快照并返回 SERVICE_NOT_READY，不再把降级清理期间形成的部分配置标记为 OK。

Keep log publishing active when optimizer config is invalid

When only the optimizer subsection is invalid, such as {"optimizer":{"queue_size":0}}, FromJsonString returns false and this early return skips registration of the otherwise default-enabled log publisher as well. A typo while enabling the optional optimizer stream therefore silently disables the existing event log; parse or validate the two publisher configurations independently so an invalid optional publisher cannot take down the other one.

Useful? React with 👍 / 👎.

Cancel blocked stream writes during demotion

When a subscribed optimizer stops reading long enough to exhaust HTTP/2 flow control, this synchronous Write can block inside the handler. DisableSubscriptions() only closes the subscription queue, so demotion cannot wake a handler already inside Write, leaving the old-leader RPC and its server thread alive until the peer reads or cancels; fresh evidence beyond the previous leadership fix is that the new close path is only observed by WaitNext, not while this call is blocked. Track and cancel active server contexts, or use a cancellable asynchronous stream, when subscriptions are disabled.

Useful? React with 👍 / 👎.

Bound subscriber buffers by serialized bytes

When a slow subscriber receives large but valid traces, this deep-copies each TraceQueryRequest into a queue limited only by message count. Because a request can contain a large block_keys array and trace ID, the default 10,000 entries can retain many gigabytes per subscriber, multiplied by up to four subscribers, allowing optimizer backpressure to exhaust KVCM memory before the count limit starts dropping events. Enforce an aggregate byte limit or a maximum serialized event size, rather than relying only on queue length.

Useful? React with 👍 / 👎.

Capture access timestamps before executing the lookup

For a slow metadata or backend lookup, this forwards a completion-time timestamp rather than the time the cache access occurred: each producing CacheManager method constructs the CacheGetEvent and calls SetEventTriggerTime() only after its lookup has returned. Fresh evidence beyond the previous timestamp discussions is this producer-side capture point, which shifts every delayed access forward by its request latency and therefore changes TTL expiration and refresh decisions even if the optimizer correctly consumes timestamp_ns. Capture the request or lookup start time and carry that value instead.

Useful? React with 👍 / 👎.